package tul.cot.job.output;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;

public class FSGetter extends OutputGetter {

	private static final PathFilter FILETR = new PathFilter() {
		
		@Override
		public boolean accept(Path path) {
			return path.getName().startsWith("part");
		}
	};
	
	public FSGetter(Configuration conf) {
		super(conf);
	}
	
	@Override
	protected Iterable<String> getLines() throws IOException {
		List<String> lines = new ArrayList<String>();
		
		FileSystem fs = FileSystem.get(getConf());
		Path fromDir = new Path(getFromDir().toString());
		Path[] files = FileUtil.stat2Paths(fs.listStatus(fromDir, FILETR));
		
		for (Path f: files) {
			BufferedReader reader = new BufferedReader(new InputStreamReader(fs.open(f)));
			String line = null;
			while ((line = reader.readLine()) != null) {
				lines.add(line);
			}
		}
		
		Collections.sort(lines, getLabelComparator());
		return lines;
	}

}
